package com.atbeijing.handler

import java.text.SimpleDateFormat
import java.util
import java.util.Date

import com.alibaba.fastjson.JSON
import com.atbeijing.bean.StartUpLog
import com.atbeijing.utils.RedisUtil
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import redis.clients.jedis.Jedis

/**
 * 打的是组合拳,driver告诉executor你该干什么,把需要的资源给你
 */
object D1 {

  def writeRedis(gDS: DStream[StartUpLog]) = {
    gDS.foreachRDD(rdd => {
      rdd.foreachPartition(partition =>{
        val client: Jedis = RedisUtil.getJedisClient
        partition.foreach(log =>{
          client.sadd("DAU"+log.logDate,log.mid)
        })
        client.close()
      })
    })
  }


  def groupDS(rDS: DStream[StartUpLog]) = {
    //我要处理各个分区
    val logDs: DStream[(String, StartUpLog)] = rDS.mapPartitions(partition => {
      //每个分区该怎么办
      partition.map(log => {
        (log.mid + log.logDate, log)
      })
    })
    val gDS: DStream[(String, Iterable[StartUpLog])] = logDs.groupByKey()
    val value: DStream[(String, List[StartUpLog])] = gDS.mapValues(iter => {
      iter.toList.sortWith(_.ts < _.ts).take(1)
    })
    value.flatMap(_._2)
  }


  def filterByRedis(ssc: StreamingContext, objDS: DStream[StartUpLog]) = {
    //整个rdd该怎么办
    objDS.transform(rdd =>{
      val sdf = new SimpleDateFormat("yyyy-MM-dd")
      val time: String = sdf.format(new Date(System.currentTimeMillis()))
      val client: Jedis = RedisUtil.getJedisClient
      val mids: util.Set[String] = client.smembers("DAU"+time)
      val midsBC: Broadcast[util.Set[String]] = ssc.sparkContext.broadcast(mids)
      client.close()
      rdd.filter(log =>{
        !midsBC.value.contains(log)
      })
    })
  }


  def transformCase(kafkaDS: InputDStream[ConsumerRecord[String, String]]) = {
    val sdf = new SimpleDateFormat("yyyy-MM-dd HH")
    //我要处理各个分区
    kafkaDS.mapPartitions(partition => {
      //每个分区该怎么办
      partition.map(record => {
        val log: StartUpLog = JSON.parseObject(record.value(), classOf[StartUpLog])
        val time: Array[String] = sdf.format(log.ts).split(" ")
        log.logDate = time(0)
        log.logHour = time(1)
        log
      })
    })
  }




}
